package com.liao.search.service.impl;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.aggregations.StringTermsAggregate;
import co.elastic.clients.elasticsearch._types.aggregations.StringTermsBucket;
import co.elastic.clients.elasticsearch.core.BulkRequest;
import co.elastic.clients.elasticsearch.core.BulkResponse;
import co.elastic.clients.elasticsearch.core.SearchRequest;
import co.elastic.clients.elasticsearch.core.SearchResponse;
import co.elastic.clients.elasticsearch.core.bulk.BulkResponseItem;
import co.elastic.clients.elasticsearch.indices.AnalyzeRequest;
import co.elastic.clients.elasticsearch.indices.AnalyzeResponse;
import co.elastic.clients.elasticsearch.indices.analyze.AnalyzeToken;
import com.liao.common.doc.HistoryKeywordDoc;
import com.liao.search.service.HistoryKeywordSaveSaveService;
import com.liao.search.utils.RedisUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

import static com.liao.common.constant.EsMappingEnum.HISTORY_KEYWORD_INDEX;
import static com.liao.common.constant.redisEnum.PRODUCT_HISTORYKEYWORD_COUNTE_KEY;

/**
 * @Author: ljt
 * @DateTime: 2025/3/13 上午10:20
 **/
@Slf4j
@Service
public class HistoryKeywordSaveSaveServiceImpl implements HistoryKeywordSaveSaveService {

    @Autowired
    private ElasticsearchClient elasticsearchClient;

    @Autowired
    private RedisUtils redisUtils;

    @Override
    public boolean historyKeyStatusUp(List<HistoryKeywordDoc> historyKeywordDocs) throws IOException {
        BulkRequest.Builder br = new BulkRequest.Builder();

        for (HistoryKeywordDoc doc : historyKeywordDocs) {
            br.operations(op -> op
                    .index(idx -> idx
                            .index(HISTORY_KEYWORD_INDEX)
                            .id(doc.getId())
                            .document(doc)
                    )
            );
        }

        BulkResponse result = elasticsearchClient.bulk(br.build());

        // Log errors, if any
        if (result.errors()) {
            log.error("历史关键词上传错误关键词{},用户id{}", historyKeywordDocs, historyKeywordDocs.get(0).getMemberId());
            for (BulkResponseItem item : result.items()) {
                if (item.error() != null) {
                    log.error(item.error().reason());
                }
            }
            return false;

        }

        return true;
    }


    @Override
    public String getMemberAlwaysKey(Integer memberId) throws IOException {
        // 2. 构建查询请求
        SearchRequest searchRequest = SearchRequest.of(s -> s
                .index(HISTORY_KEYWORD_INDEX)
                .size(0) // 不返回文档
                .query(q -> q
                        .bool(b -> b
                                .filter(f -> f
                                        .term(t -> t
                                                .field("member_id")
                                                .value(memberId)
                                        )
                                )
                        )
                )
                .aggregations("topN", a -> a
                        .terms(t -> t
                                .field("member_keyword")

                                .size(10)
                        )
                )
        );

        // 3. 执行查询
        SearchResponse<Void> response = elasticsearchClient.search(searchRequest, Void.class);

        // 4. 解析聚合结果
        StringTermsAggregate terms = response.aggregations()
                .get("topN")
                .sterms();

        StringBuilder sb = new StringBuilder();

        for (StringTermsBucket bucket : terms.buckets().array()) {

//            sb.append(bucket.key()._get().toString()).append("  ");

        }

        System.out.println("关键词" + sb.toString());
        return sb.toString();
    }

    @Override
    public boolean save(String text, Integer MenmberId) throws IOException {

        List<HistoryKeywordDoc> historyKeywordDocList = new ArrayList<>();
        List<String> keyWords = analyzeText(text);
//        获取当前对象的计数器值
        Integer counter = redisUtils.get(PRODUCT_HISTORYKEYWORD_COUNTE_KEY + MenmberId) == null ? 0 : Integer.parseInt(redisUtils.get(PRODUCT_HISTORYKEYWORD_COUNTE_KEY + MenmberId));

        for (String keyword : keyWords) {
            counter++;
            // 将计数器值+1
            redisUtils.set(PRODUCT_HISTORYKEYWORD_COUNTE_KEY + MenmberId, counter, RedisUtils.NOT_EXPIRE);
            // 将用户id和关键词存入es的历史关键词中
            historyKeywordDocList.add(new HistoryKeywordDoc(counter.toString(), MenmberId.toString(), keyword));
        }
        //保存关键词数据
        historyKeyStatusUp(historyKeywordDocList);
        return true;
    }

    /**
     * 获取用户访问文本转化成关键词
     *
     * @param text
     * @return
     * @throws IOException
     */
    public List<String> analyzeText(String text)
            throws IOException {

        // 构建分析请求
        AnalyzeRequest request = AnalyzeRequest.of(a -> a
                .analyzer("ik_smart")
                .text(text)
        );

        // 执行请求
        AnalyzeResponse response = elasticsearchClient.indices().analyze(request);

        // 提取分词结果
        return response.tokens().stream()
                .map(AnalyzeToken::token)
                .collect(Collectors.toList());
    }

}
